package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class Flink01_Batch_WordCount {
    public static void main(String[] args) throws Exception {
        /**
         * spark中WordCount过程
         * 1.创建spark环境
         * 2.通过调用读文件的方法读取文件中的数据
         * 3.调用flatMap方法，对数据按照空格切分
         * 4.调用map方法，对切出来的数据组成Tuple2元组<word,1>
         * 5.调用了reduceBykey方法，对相同单词的数据的value值做累加
         *          |
         *          |---->1.将相同单词的数据聚合到一块 类似于sql中groupBy
         *          |---->2.对value值做累加 类似于sql中sum（）
         * 6.打印到控制台
         * 7.关闭资源
         */

        //1.获取Flink批处理执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //2.通过调用读文件的方法读取文件中的数据
        DataSource<String> dataSource = env.readTextFile("input/word.txt");

        //3.调用flatMap方法，对数据按照空格切分,返回切出来的每一个单词
        FlatMapOperator<String, String> wordDataSource = dataSource.flatMap(new MyFlatMap());

        //4.调用map方法，对切出来的数据组成Tuple2元组<word,1>
        MapOperator<String, Tuple2<String, Integer>> wordToOneDataSource = wordDataSource.map(new MapFunction<String, Tuple2<String, Integer>>() {
            /**
             *
             * @param value 输入的数据
             * @return
             * @throws Exception
             */
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
//                return new Tuple2<>(value, 1);
                return Tuple2.of(value, 1);
            }
        });


        //5.将相同单词的数据聚合到一块 类似于sql中groupBy
        UnsortedGrouping<Tuple2<String, Integer>> groupBy = wordToOneDataSource.groupBy("f0");

        //6.对value值做累加 类似于sql中sum（）
        AggregateOperator<Tuple2<String, Integer>> result = groupBy.sum(1);

        result.print();

    }

    public static class MyFlatMap implements FlatMapFunction<String,String>{

        /**
         *
         * @param value 输入的数据
         * @param out 收集器，可以把数据发送至下游
         * @throws Exception
         */
        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            //对数据按照空格切分
            String[] words = value.split(" ");
            //遍历字符串数据数组获取到每一个单词
            for (String word : words) {
                out.collect(word);
            }
        }
    }
}
